package midas.mw.crt;

import midas.mw.cfg.CfgError;
import midas.mw.cfg.CfgLogging;
import midas.mw.cfg.log.Logger;
import midas.mw.crt.MD.PriorityTable;
import midas.mw.crt.common.IPMessage;
import midas.mw.crt.interfaces.CrtCallback;

/**
 * This class manages the thread responsible of sending messages
 */
public class QueueThread extends Thread
{
	/**
	 * Priority table associated to this QueueThread
	 */
	PriorityTable priorityTable = null;
	
	String nodeId = null;
	
        /**
	 * Instance of the CRT running in the node
	 */
	CrtImpl crt = null;
	
	/**
	 * Time to wait for a response
	 */
	private long waitingTime 		= 30000;
	
	/**
	 * String used to provide locks
	 */
	public String responseLock 		= new String("responseLock");
	
	/**
	 * Shows if a response has been got or not
	 */
	public boolean bResolved		= false;	
		
	/**
	 * Variable used to know if there is more messages ready to be sent
	 */	
	boolean thereIsMoreMessages		= true;
	
	/**
	 * Constructor - initiate local variables
	 * @param crt Instance of the CRT running in the node
	 * @param nodeId MIDAS node identifier
	 */
	public QueueThread (CrtImpl crt, String nodeId)
	{		
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "QueueThread(CrtImpl, " + nodeId + ")");		
		
		this.priorityTable = (PriorityTable)crt.getMessageDispatcher().getNodeList().getNodeQueue(nodeId);
		this.nodeId 		= nodeId;
		this.crt 			= crt;		
	}
	
	/**
	 * Overriding the run method of class Thread
	 */
	public void run()
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "QueueThread.run()");		
        
		IPMessage messageStored = null;
		
		while (thereIsMoreMessages)
		{
			//if (priorityTable.equals(crt.getOwnNodeId()))
			//	return;				
			//while (!crt.getNodeIdentifierSolver().getNodeIdToIpConnectedPeers().containsKey(nodeId))
			//wait();
			messageStored = null;
			
			//Podemos intentar mandar el mensaje
			synchronized (priorityTable)
			{
				if (!priorityTable.isEmptyTable())				
					messageStored = (IPMessage)priorityTable.seeFirstToBeSent();								
				else				
					thereIsMoreMessages = false;								
			}
			
			if (messageStored != null)
			{			
				//Se puede tratar de un mensaje multiple, por lo que le descomponemos en tantos mensajes
				//como sea necesario
				try {
					//Dependiendo del tipo de mensaje
					//Si es un mensaje corto o ya ha sido dividido porque viene de un node JXTA o JXME al proxy
					if (messageStored.getNumMessages() != 0)
						sendEnqueuedMessage(messageStored, crt.getNodeIdentifierSolver().isJXMENode(nodeId));
					else
						sendChunkedMessage(messageStored, crt.getNodeIdentifierSolver().isJXMENode(nodeId));
					
					//Si el origen del mensaje no es el propio nodo borramos el mensaje
					if (messageStored.getConfirmation() == false || !messageStored.getSourceNodeId().equals(crt.getOwnNodeId()))
					{
						crt.getMessageDispatcher().getMessageList().remove(messageStored.getMessageId());
						crt.getMessageDispatcher().getTimeQueue().deleteMessage(messageStored);
						synchronized (priorityTable)
						{
							priorityTable.deleteMessage(messageStored);
							if (priorityTable.isEmptyTable())					
								crt.getMessageDispatcher().getNodeList().getNodes().remove(messageStored.getDestinationNodeId());									
						}				
					}
					else if (messageStored.getMessageType().equals(Constants.IP_MESSAGE_REQUEST) && messageStored.getConfirmation()
							&& messageStored.getSourceNodeId().equals(crt.getOwnNodeId()))
							//&& messageStored.getNumOrder() == messageStored.getNumMessages())
					{
						synchronized (responseLock) 
						{
							bResolved = false;
							
								try {
								CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "QueueThread.run(). Wainting answer ....");		
							    
								responseLock.wait(waitingTime);
								/*if (bResolved)	
										System.out.println("SE RECIBE RESPUESTA PARA: " + messageStored.getMessageId());
									else
										System.out.println("NO SE HA RECIBIDO RESPUESTA");
								*/
								if (bResolved)
								{		
									crt.getMessageDispatcher().getMessageList().remove(messageStored.getMessageId());
									crt.getMessageDispatcher().getTimeQueue().deleteMessage(messageStored);
									synchronized (priorityTable)
									{
										priorityTable.deleteMessage(messageStored);
										if (priorityTable.isEmptyTable())					
											crt.getMessageDispatcher().getNodeList().getNodes().remove(messageStored.getDestinationNodeId());									
									}	
								}
								else
								{
									CfgLogging.logEvent(CfgLogging.LOG_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "QueueThread.run(). There is not answer");		
							        
									//First, we look for the destination among the registered components (but not the MW)
									CrtCallback crtUser = (CrtCallback) crt.getCRTUser(messageStored.getSourceUserId());
									if (crtUser != null)
										crtUser.processUndeliveredMessage(messageStored);
									else
									{
										//We look for the destination between components
										try {
											crtUser = (CrtCallback) crt.getMiddlewareInterface().getRunningComponent(messageStored.getSourceUserId());
										} catch (ClassCastException cce){}
										if (crtUser != null)
											crtUser.processUndeliveredMessage(messageStored);
										//We look for the destination between MIDAS applications					
										else
											CfgError.errorReporting(CfgError.ERR_INFO, Logger.COMPONENT_CRT, CfgLogging.SEV_INFORMAL, "MessageDispatcher.messagesExpired(List). " + 
													"The source application is not registered", null);				
									}
							}
							} catch (InterruptedException e) {}
						}
					}
				} catch (CrtException crte) {					
					if (crte.value == 1)						
						return;
					else
					{
						//if (messageStored.getConfirmation() == false || !messageStored.getSourceNodeId().equals(crt.getOwnNodeId()))
						//{
						//System.out.println("DEBERIAMOS BORRAR EL MENSAJE");
						crt.getMessageDispatcher().getMessageList().remove(messageStored.getMessageId());
							crt.getMessageDispatcher().getTimeQueue().deleteMessage(messageStored);
							synchronized (priorityTable)
							{
								priorityTable.deleteMessage(messageStored);
								if (priorityTable.isEmptyTable())					
									crt.getMessageDispatcher().getNodeList().getNodes().remove(messageStored.getDestinationNodeId());									
							}										
					}						
				}			
			}			
		}								
	}	
	
	private void sendChunkedMessage(IPMessage messageStored, boolean isJXME) throws CrtException
	{
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.sendChunkedMessage(" + messageStored.msgToString() + ", " + isJXME + ")");
		
		try {
			int totalLength = messageStored.getTotalLength();								
			int numCompleteMessges = totalLength/Constants.MAX_MESS_SIZE;
			int rest = totalLength%Constants.MAX_MESS_SIZE;			
			int numBytes = 0;
			int contador = 1;						
			int numMessages = (rest == 0) ? numCompleteMessges : (numCompleteMessges + 1);
			messageStored.setNumMessages(numMessages);			
			while (contador<=numCompleteMessges)
			{													
				messageStored.setNumOrder(contador);				
				
				if (isJXME)
					crt.getJxtaPlatform().forwardMessage(messageStored, numBytes, Constants.MAX_MESS_SIZE);
				else
					crt.getPacketForwarder().sendMessage(messageStored, numBytes, Constants.MAX_MESS_SIZE);														
				numBytes+=Constants.MAX_MESS_SIZE;
				contador++;	
				try {	
					Thread.sleep(300);
				} catch(InterruptedException ie) {				
					ie.printStackTrace();
				}
			}
											
			if (rest > 0)
			{																													
				messageStored.setNumOrder(contador);				
								
				//crt.getPacketForwarder().sendMessage(messageCloned);
				if (isJXME)
					crt.getJxtaPlatform().forwardMessage(messageStored, numBytes, rest);
				else
					crt.getPacketForwarder().sendMessage(messageStored, numBytes, rest);
				
				try {	
					Thread.sleep(300);
				} catch(InterruptedException ie) {				
					ie.printStackTrace();
				}
				
			}																				
		} catch (CrtException crte){
			throw crte;
		}
	}
	
	private void sendEnqueuedMessage(IPMessage messageStored, boolean isJXME) throws CrtException
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "MessageDispatcher.sendEnqueuedMessage(" + messageStored.msgToString() + ", " + isJXME + ")");
		
		try {			
			if (isJXME)
				crt.getJxtaPlatform().forwardMessage(messageStored);
			else
				crt.getPacketForwarder().sendMessage(messageStored);																							
		} catch (CrtException crte){
			throw crte;
		}
	}
	
	public void responseGot()
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "QueueThread.responseGot()");	
		
		bResolved = true;
		synchronized(responseLock) {
			responseLock.notify();
		}
	}
	
	public void stopWaiting()
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "QueueThread.stopWaiting()");	
		
		synchronized(responseLock) {
			responseLock.notify();
		}
	}
	
	public void stopThread()
	{	
		CfgLogging.logEvent(CfgLogging.LOG_MCALL, Logger.COMPONENT_CRT, CfgLogging.SEV_DEBUG, "QueueThread.stopThread()");	
		
		thereIsMoreMessages = false;
	}
}
